// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/feed/mgr/system.h"

#include <map>
#include <memory>
#include <string>
#include <tuple>
#include <utility>
#include <vector>
#include <unordered_set>

#include "coin2/base/config/app_config.h"
#include "coin2/exchange/base/executor/archive_executor.h"
#include "coin2/exchange/base/executor/live_executor.h"
#include "coin2/exchange/base/executor_util/market_ts.h"
#include "coin2/exchange/base/feed/processor.h"
#include "coin2/exchange/base/market/enums.h"
#include "coin2/exchange/base/market/market_exchange_api.h"

// feed
#include "coin2/exchange/base/feed/live/system.h"
#include "coin2/exchange/base/feed/live_parallel/system.h"
#include "coin2/exchange/base/feed/live_writer/system.h"
#include "coin2/exchange/base/feed/system.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/feed/archive/system.h"
#include "coin2/feed/fastfeed/system.h"
#include "coin2/feed/universe/system.h"
#include "coin2/feed/navi/system.h"
#include "coin2/feed/relay/system.h"
#include "coin2/exchange/base/feed/timer/timer_manager.h"

namespace coin2::feed::mgr {
namespace impl {

IFeedSubscriber* FeedSubsystemManager::InitTimerManager(
    const FeedManagerConfig& config,
    IFeedSubscriber* strategy,
    boost::asio::io_context& main_ioc) {
  // no feed exchagnes specified
  if (config.exchanges_size() == 0) {
    return strategy;
  }

  FeedSystemType type = FeedSystemType::FEED_SYSTEM_TYPE_UNSPECIFIED;
  if (config.has_common() && config.common().has_type()) {
    type = config.common().type();
  } else if (config_.exchanges().size() > 0) {
    type = config_.exchanges().begin()->second.type();
  } else {
    NOTREACHED();
  }

  if (type == FeedSystemType::FEED_SYSTEM_TYPE_LIVE) {
    auto timer = coin2::exchange::base::feed::timer::TimerManager::CreateLiveManager();
    timer->RegisterSubscriber(strategy);
    timer->RegisterTimerIoc(main_ioc);
    return strategy;
  } else if (type == FeedSystemType::FEED_SYSTEM_TYPE_ARCHIVE) {
    auto timer = coin2::exchange::base::feed::timer::TimerManager::CreateArchiveManager();
    return timer->HookSubscriber(strategy);
  }

  return strategy;
}

void FeedSubsystemManager::InitCounter(
    const FeedManagerConfig& config) {
  // no feed exchagnes specified
  if (config.exchanges_size() == 0) {
    return ;
  }

  bool enable_counter = false;
  if (config.has_common() && config.common().has_type()) {
    if (config.common().type() != FeedSystemType::FEED_SYSTEM_TYPE_LIVE) return;
    enable_counter = config.common().message_counter();
  } else if (config.exchanges().size() > 0) {
    if (config.exchanges().begin()->second.type() != FeedSystemType::FEED_SYSTEM_TYPE_LIVE) return;
    enable_counter = config_.exchanges().begin()->second.message_counter();
  }

  if (enable_counter) {
    feed_counter_ = std::make_unique<FeedCounter>();
    lass_->EnableCounter(feed_counter_.get());
  }
}

FeedSubsystemManager::FeedSubsystemManager(
    const coin2::exchange::di::ConstantInjection& ci,
    const FeedManagerConfig& config,
    LiveAsyncExecutor* lass,
    ArchiveSyncExecutor* aass,
    exchange::base::feed::IFeedSubscriber* strategy)
    : config_(config), lass_(lass), aass_(aass) {
  auto hook_strategy = InitTimerManager(config_, strategy, *(lass->MainIoc()));
  InitCounter(config);
  for (const auto& [system_name, system] : config_.exchanges()) {
    auto mea = MarketExchangeApi::FromString(system.mea());
    auto recipe = system.recipe();
    auto mear = MarketExchangeApiRecipe{mea, recipe};
    auto type = system.type();
    feed_system_types_.push_back(type);

    if (type == FeedSystemType::FEED_SYSTEM_TYPE_UNSPECIFIED) {
      THROW() << "[FeedSubsystemManager] Please configure system type for " << system.mea();
    } else if (type == FeedSystemType::FEED_SYSTEM_TYPE_LIVE) {
      CHECK_THROW(lass_ != nullptr) << "You didn't provide LiveAsyncExecutor";
      if (lass->ParserIoc() != nullptr) {
        systems_.push_back(std::make_unique<exchange::base::feed::LiveParallelFeedSubsystem>(
            system_name,
            mear.mea,
            ci,
            system,
            lass_,
            strategy,
            feed_counter_.get()));
      } else {
        systems_.push_back(std::make_unique<exchange::base::feed::LiveFeedSubsystem>(
            system_name,
            mear.mea,
            ci,
            system,
            lass_,
            strategy));
      }
    } else if (type == FeedSystemType::FEED_SYSTEM_TYPE_LIVE_WRITER) {
      if (system.replay_only()) continue;
      // provide mea, not mear.
      CHECK_THROW(lass_ != nullptr) << "You didn't provide LiveFeedWriterSubsystem";
      systems_.push_back(std::make_unique<exchange::base::feed::LiveFeedWriterSubsystem>(
          system_name,
          mea,
          ci,
          system,
          lass_));
    } else if (
        type == FeedSystemType::FEED_SYSTEM_TYPE_ARCHIVE &&
        system.archive().archive_type() ==
            ArchiveConfig_ArchiveType::ArchiveConfig_ArchiveType_RAW) {
      CHECK_THROW(aass_ != nullptr) << "You didn't provide ArchiveSyncExecutor";
      systems_.push_back(std::make_unique<archive::RawArchiveFeedSystem>(
          system_name,
          mear.mea,
          ci,
          system,
          aass_,
          hook_strategy));
    } else if (
        type == FeedSystemType::FEED_SYSTEM_TYPE_ARCHIVE &&
        system.archive().archive_type() ==
            ArchiveConfig_ArchiveType::ArchiveConfig_ArchiveType_FASTFEED) {
      CHECK_THROW(aass_ != nullptr) << "You didn't provide ArchiveSyncExecutor";
      systems_.push_back(std::make_unique<fastfeed::FastFeedSubsystem>(
          system_name,
          mear,
          ci,
          system,
          aass_,
          hook_strategy));
    } else if (
        type == FeedSystemType::FEED_SYSTEM_TYPE_ARCHIVE &&
        system.archive().archive_type() ==
            ArchiveConfig_ArchiveType::ArchiveConfig_ArchiveType_FASTFEED_UNIVERSE) {
      CHECK_THROW(aass_ != nullptr) << "You didn't provide ArchiveSyncExecutor";
      systems_.push_back(std::make_unique<fastfeed::FastFeedUniverseSubsystem>(
          system_name,
          mear,
          ci,
          system,
          aass_,
          hook_strategy));
    } else if (type == FeedSystemType::FEED_SYSTEM_TYPE_NAVI_ARCHIVE) {
      if (system.archive().archive_type() ==
          ArchiveConfig_ArchiveType::ArchiveConfig_ArchiveType_RAW) {
      } else if (system.archive().archive_type() ==
                 ArchiveConfig_ArchiveType::
                     ArchiveConfig_ArchiveType_FASTFEED) {
        CHECK_THROW(aass_ != nullptr) << "You didn't provide ArchiveSyncExecutor";
        systems_.push_back(std::make_unique<NaviFeedArchiveSystem>(
            system_name, mear.mea, ci, system, aass_, strategy));                      
      }
    } else if (type == FeedSystemType::FEED_SYSTEM_TYPE_RELAY) {
      CHECK_THROW(lass_ != nullptr) << "You didn't provide ArchiveSyncExecutor";
      systems_.push_back(std::make_unique<relay::FeedRelaySubsystem>(
          system_name,
          mear,
          ci,
          system,
          lass_,
          strategy));
    } else {
      throw std::runtime_error("unexpected feed system type");
    }
  }
  strategy->InjectFeed(this);
}

FeedSubsystemManager::~FeedSubsystemManager() {
  coin2::exchange::base::feed::timer::TimerManager::DeleteInstance();
}

std::vector<IFeedSubsystem*> FeedSubsystemManager::GetAllFeedSubsystems() {
  std::vector<IFeedSubsystem*> ret;
  for (auto& s : systems_) {
    ret.push_back(s.get());
  }
  return ret;
}

std::vector<IProduct*> FeedSubsystemManager::GetProducts() const {
  std::vector<IProduct*> vec;
  std::unordered_set<std::string> duplicate_check;
  for (const auto& system : systems_) {
    for (auto* product : system->GetProducts()) {
      std::string key = product->relative_norm() + product->absolute_norm();
      if (duplicate_check.count(key) == 0) {
        vec.push_back(product);
        duplicate_check.insert(key);
      }
    }
  }
  return vec;
}

void FeedSubsystemManager::DebugPrint() const {
  LOG(INFO) << "[FeedManager]";
  if (lass_ != nullptr) {
    LOG(INFO) << "\t   live ass = <..>";
  } else {
    LOG(INFO) << "\t   live ass = nullptr";
  }
  if (aass_ != nullptr) {
    LOG(INFO) << "\tarchive ass = <..>";
  } else {
    LOG(INFO) << "\tarchive ass = nullptr";
  }

  for (auto& s : systems_) {
    s->DebugPrint();
  }
}

void FeedSubsystemManager::GetHealthReport(int indent, std::string* report) const {
  // [HealthReport]
  //     [Feed]
  //         - Huobi (OK)
  //         - Okex (STARTING, uptime=23h)
  //               a1 (..)
  for (int i = 0; i < indent * 4; i++) {
    *report += ' ';
  }
  if (this->GetCounter()) {
    std::stringstream ss;
    ss << "[Feed (received=" << this->GetCounter()->ReceivedNum()
       << ", processed=" << this->GetCounter()->ProcessedNum()
       << ", published=" << this->GetCounter()->PublishedNum()
       << ", consumed=" << this->GetCounter()->ConsumedNum()
       << ")]\n";
    *report += ss.str();
  } else {
    *report += "[Feed]\n";
  }
  for (auto& system : systems_) {
    system->GetHealthReport(indent + 1, report);
  }
}

}  // namespace impl

}  // namespace coin2::feed::mgr
